package com.atguigu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.app.func.MyAllWindowFunction;
import com.atguigu.bean.UserNew;
import com.atguigu.utils.MyClickHouseUtil;
import com.atguigu.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

import java.time.Duration;

public class Dws0202UserNewWindow_00 {
    public static void main(String[] args) throws Exception {
        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
//        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(18*60+5));
        //  状态后端设置
//        env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L);
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
//        env.getCheckpointConfig().enableExternalizedCheckpoints(
//                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
//        );
//        env.setRestartStrategy(RestartStrategies.failureRateRestart(
//                10, Time.of(1L, TimeUnit.DAYS), Time.of(3L, TimeUnit.MINUTES)
//        ));
//        env.setStateBackend(new HashMapStateBackend());
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
//        System.setProperty("HADOOP_USER_NAME", "atguigu");

        //2.从kafka DWD层 读取dwd_user_register主题的数据并生成watermark
        DataStreamSource<String> kafkaDS = env.fromSource(MyKafkaUtil.getKafkaSource("dwd_user_register", "user_new"), WatermarkStrategy.<String>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<String>() {
            @Override
            public long extractTimestamp(String s, long l) {
                JSONObject jsonObject = JSON.parseObject(s);
                return jsonObject.getLong("create_time");
            }
        }), "kafka-source");
        kafkaDS.print("kafkaDS==>");


        //3.将数据转换为JavaBean对象
        SingleOutputStreamOperator<UserNew> userActiveDS = kafkaDS.map(line ->
                new UserNew("", "", 1L, 0L)
        );
        userActiveDS.print("userActiveDS-->");

        //4.开窗 聚合
        SingleOutputStreamOperator<UserNew> reduceDS = userActiveDS.windowAll(TumblingEventTimeWindows.of(Time.seconds(10))).reduce(new ReduceFunction<UserNew>() {
            @Override
            public UserNew reduce(UserNew userNewActive, UserNew t1) throws Exception {
                userNewActive.setNewUserCount(userNewActive.getNewUserCount()+t1.getNewUserCount());

                System.out.println("chen");
                return userNewActive;
            }
        }, new MyAllWindowFunction<UserNew, TimeWindow>());

        //写出
        reduceDS.print("reduceDS--->");
        reduceDS.addSink(MyClickHouseUtil.getSinkFunction("insert into dws_user_new_window values (?,?,?,?)"));


        env.execute();


    }
}
